package com.hong.qrpc.handler.in.response;

import com.hong.qrpc.constant.ProtoTypeConst;
import com.hong.qrpc.constant.QrpcConst;
import com.hong.qrpc.factory.ResolverFactory;
import com.hong.qrpc.header.QrpcHeader;
import com.hong.qrpc.resolver.Resolver;
import com.hong.qrpc.task.NettyTask;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author HQT
 * @Date 2022/8/10 15:33
 */
@Slf4j
public class QrpcResponseHandler extends ChannelInboundHandlerAdapter {
    private ReentrantLock reentrantLock = new ReentrantLock();
    private ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
    private ConcurrentHashMap<Long, NettyTask<InputStream>> resultMap;
    private int protoType;
    private boolean isReadHeader;
    private QrpcHeader header;

    public QrpcResponseHandler(ConcurrentHashMap<Long, NettyTask<InputStream>> resultMap, int protoType) {
        this.resultMap = resultMap;
        this.protoType = protoType;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //先上锁，然后再逐个解析响应并且唤醒等待响应的线程
        reentrantLock.lock();
        try {
            byteBuf.writeBytes((ByteBuf) msg);
            ((ByteBuf) msg).release();
            while (isFullRequest(byteBuf)) {
                log.debug("response header:{}", header);
                byte[] temp = new byte[header.getDataLen()];
                byteBuf.readBytes(temp);
                NettyTask<InputStream> nettyTask = resultMap.remove(header.getRequestId());
                nettyTask.putAndNotify(new ByteArrayInputStream(temp));
            }
        } finally {
            reentrantLock.unlock();
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }

    private boolean isFullRequest(ByteBuf byteBuf) {
        Resolver resolver = ResolverFactory.getResolver(ProtoTypeConst.QRPC);
        if (isReadHeader) {
            //已经读了包头了，数据包没有读完整的情况
            if (byteBuf.readableBytes() >= header.getDataLen()) {
                //因为这个包在下次调用之前就读完了，下次就从头开始读起
                isReadHeader = false;
                return true;
            } else {
                return false;
            }
        } else {
            if (!(byteBuf.readableBytes() >= QrpcConst.HEADER_BYTE_LEN)) {
                return false;
            }
            header = (QrpcHeader) resolver.readHeader(this.byteBuf);
            isReadHeader = true;
            if (byteBuf.readableBytes() >= header.getDataLen()) {
                //因为这个包在下次调用之前就读完了，下次就从头开始读起
                isReadHeader = false;
                return true;
            } else {
                return false;
            }
        }
    }
}
